package com.atguigu.tingshu.album.mq;


import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.service.TrackInfoService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.concurrent.TimeUnit;


@Component
@Slf4j
public class TrackReceiver {


    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private TrackInfoService trackInfoService;


    @KafkaListener(topics = KafkaConstant.QUEUE_TRACK_STAT_UPDATE)
    public void updateStat(ConsumerRecord<String, String> record) {
        String value = record.value();
        if (!StringUtils.isEmpty(value)) {
            TrackStatMqVo trackStatMqVo = JSON.parseObject(value, TrackStatMqVo.class);
            String businessNo = trackStatMqVo.getBusinessNo();
            String redisKey = "track:stat:" + businessNo;
            Boolean setIfAbsent = redisTemplate.opsForValue().setIfAbsent(redisKey, businessNo, 1, TimeUnit.HOURS);

            if (setIfAbsent) {
                //如果setIfAbsent()返回true，说明当前是首次处理该businessNo的统计信息
                trackInfoService.trackStatUpdate(trackStatMqVo);
            }
        }
    }
}
